package net.techfuser.bigscreen;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import net.techfuser.bigscreen.entity.ItemSummary;
import net.techfuser.bigscreen.entity.Summary;
import net.techfuser.bigscreen.sink.ItemSummayRedisMapper;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.triggers.ContinuousProcessingTimeTrigger;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.streaming.connectors.redis.RedisSink;
import org.apache.flink.streaming.connectors.redis.common.config.FlinkJedisPoolConfig;
import org.apache.flink.util.Collector;

import java.text.SimpleDateFormat;
import java.time.Duration;
import java.util.Properties;

/**
 * @Description:
 * @Author 左龙龙
 * @Date 21-4-6
 * @Version 1.0
 **/
public class OrderDetailTask {

    public static void main(String[] args) throws Exception {
        // 0.env
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);

        // 1.source
        FlinkKafkaConsumer<String> flinkKafkaConsumer = getFlinkKafkaConsumer();
//        flinkKafkaConsumer.setStartFromLatest();
//        flinkKafkaConsumer.setStartFromEarliest();
        flinkKafkaConsumer.setStartFromTimestamp(1618844400000L);
        DataStreamSource<String> kafkaDS = env.addSource(flinkKafkaConsumer);


//        kafkaSD.print("kafka order");
//        kafkaSD.writeAsText("data/kafka/order_oplog.txt", FileSystem.WriteMode.OVERWRITE).setParallelism(1);


        // 2.transformation
        SingleOutputStreamOperator<JSONObject> orderDetailDS = kafkaDS.map(JSON::parseObject)
                .filter(oplog -> "i".equals(oplog.getString("op")))
                .map(oplog -> oplog.getJSONObject("data"))
//                .filter(order -> {
//                    SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd");
//                    return "2021-04-08".equals(sdf.format(order.getDate("order_time")));
//                })
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy.<JSONObject>forBoundedOutOfOrderness(Duration.ofSeconds(0))
                                .withTimestampAssigner((order, timestamp) -> order.getDate("order_time").getTime())
                                .withIdleness(Duration.ofSeconds(1))
                )
                .name("oplog 数据过滤");

//        orderDS.print("orderDS");


        SingleOutputStreamOperator<ItemSummary> resultDS = orderDetailDS.keyBy(new KeySelector<JSONObject, Tuple3<String, String,String>>() {
            @Override
            public Tuple3<String, String,String> getKey(JSONObject row) throws Exception {
                return Tuple3.of(row.getString("item_code"), row.getString("barcode"), row.getString("item_name"));
            }
        })
                .window(TumblingEventTimeWindows.of(Time.days(1), Time.hours(-8)))
                .trigger(ContinuousProcessingTimeTrigger.of(Time.seconds(1)))
                .aggregate(new AggregateFunction<JSONObject, Summary, Summary>() {
                    @Override
                    public Summary createAccumulator() {
                        return new Summary();
                    }

                    @Override
                    public Summary add(JSONObject value, Summary accumulator) {
                        accumulator.count += value.getIntValue("sale_qty");
                        accumulator.total_sale_value += value.getIntValue("sale_value");
                        return accumulator;
                    }

                    @Override
                    public Summary getResult(Summary accumulator) {
                        return accumulator;
                    }

                    @Override
                    public Summary merge(Summary a, Summary b) {
                        a.count += b.count;
                        a.total_sale_value += b.total_sale_value;
                        return a;
                    }
                }, new WindowFunction<Summary, ItemSummary, Tuple3<String, String, String> , TimeWindow>() {
                    @Override
                    public void apply(Tuple3<String, String, String> keys, TimeWindow window, Iterable<Summary> input, Collector<ItemSummary> out) throws Exception {
                        ItemSummary itemSummary = new ItemSummary();
                        itemSummary.item_code = keys.f0;
                        itemSummary.barcode = keys.f1;
                        itemSummary.item_name = keys.f2;

                        Summary summary = input.iterator().next();
                        itemSummary.count = summary.count;
                        itemSummary.total_sale_value = summary.total_sale_value;

                        SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
                        itemSummary.window_time = String.format("[%s,%s)", sdf.format(window.getStart()), sdf.format(window.getEnd()));
                        itemSummary.timestamp = sdf.format(System.currentTimeMillis());

                        out.collect(itemSummary);
                    }
                });
//        resultDS.print("result");

        // 3.sink
        FlinkJedisPoolConfig conf = new FlinkJedisPoolConfig.Builder()
                .setHost("192.168.40.221")
                .setPassword("techfuser")
                .setDatabase(6)
                .build();
        RedisSink<ItemSummary> redisSink = new RedisSink<ItemSummary>(conf, new ItemSummayRedisMapper());
        resultDS.addSink(redisSink).name("数据结果存redis");

        // 4.execute
        env.execute("订单明细oplog消费");
    }


    private static FlinkKafkaConsumer<String> getFlinkKafkaConsumer() {
        // 准备 kafka 连接参数
        Properties properties = new Properties();
        // 集群地址
        properties.setProperty("bootstrap.servers", "172.16.10.23:9092");
//        properties.setProperty("bootstrap.servers", "127.0.0.1:9092");
        // 消费组
        properties.setProperty("group.id", "qnh-big-screen-test");
        // 有 offset 消费记录则从记录位置继续消息, 没有消费记录则从最新的（最后的）消息开始消费
//        properties.setProperty("auto.offset.reset", "latest");
//        properties.setProperty("auto.offset.reset", "earliest");
        // 会开启一个后台线程每隔5秒检查一下 kafka 的分区情况, 实现动态分区检测
        properties.setProperty("flink.partition-discovery.interval-millis", "5000");
        // 自动提交
        properties.setProperty("enable.auto.commit", "true");
        // 自动提交的时间间隔
        properties.setProperty("auto.commit.interval.ms", "2000");

        return new FlinkKafkaConsumer<>("realtime_order_detail", new SimpleStringSchema(), properties);
    }

}


